using Confluent.Kafka;
using JT808Gateway.Configs;
using JT808Protocol.Abstracts;
using JT808Protocol.Models;
using System;
using System.Net;
using System.Threading.Tasks;

namespace JT808Gateway.Forwarders
{
    /// <summary>
    /// Kafka转发器
    /// </summary>
    public class KafkaForwarder : IDataForwarder
    {
        private KafkaConfig _kafkaConfig;

        private ProducerConfig _producerConfig;

        public KafkaForwarder(KafkaConfig config)
        {
            _kafkaConfig = config;

            _producerConfig = new ProducerConfig
            {
                BootstrapServers = $"{_kafkaConfig.Host}:{_kafkaConfig.Port}",
                ClientId = Dns.GetHostName()
            };
        }

        /// <summary>
        /// 使用Kafka转发消息
        /// </summary>
        /// <param name="package"></param>
        /// <returns></returns>
        public async Task Send(JT808Package package)
        {
            using (var producer = new ProducerBuilder<string, JT808Package>(_producerConfig).Build())
            {
                try
                {
                    //Todo : 转发消息供微服务消费
                    var dr = await producer.ProduceAsync(_kafkaConfig.JT808GwSendTopic, new Message<string, JT808Package>
                    {
                        Key = $"{package.Header.MsgID}",
                        Value = package
                    });
                    Console.WriteLine($"Delivered '{dr.Value}' to '{dr.TopicPartitionOffset}'");
                }
                catch (ProduceException<Null, string> e)
                {
                    Console.WriteLine($"Delivery failed: {e.Error.Reason}");
                }
            }
        }
    }
}